package com.spx.chapter09;

import akka.stream.impl.ReducerState;
import com.spx.chapter05.pojo.Event;
import com.spx.util.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * create by undeRdoG on  2022-05-04  22:03
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class StateTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );


        stream.keyBy(data -> data.user)
                .flatMap(new RichFlatMap())
                .print();

        env.execute();

    }


    public static class RichFlatMap extends RichFlatMapFunction<Event,String> {

        private ValueState<Event> myValueState;

        private ListState<Event> myListState;

        private MapState<String,Long> myMapState;

        private ReducingState<Event> myReduceingState;

        private AggregatingState<Event,String> myArrregategingState;

        @Override
        public void open(Configuration parameters) throws Exception {
            ValueStateDescriptor<Event> myValueState = new ValueStateDescriptor<>("myValueState", Types.POJO(Event.class));
            this.myValueState = getRuntimeContext().getState(myValueState);

            // 设置 state的 ttl时间，目前默认使用的是  ProcessTime
            StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(1))
                    // 更新策略，这里设置的是当 read 和 write 都会更新ttl时间，默认的是  OnCreateAndWrite
                    .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                    // 设置 过期数据的可见性，因为过期state是懒删除，当访问到过期数据时，当前state还是存在的，ReturnExpiredIfNotCleanedUp 表示可以读到过期的state
                    .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
                    .build();

            myValueState.enableTimeToLive(ttlConfig);

            this.myListState = getRuntimeContext().getListState(new ListStateDescriptor<Event>("myListState", Event.class));

            this.myMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Long>("myMapState", String.class, Long.class));

            this.myReduceingState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Event>("myReducingState", new ReduceFunction<Event>() {
                @Override
                public Event reduce(Event value1, Event value2) throws Exception {
                    return new Event(value1.user,value1.url,value2.timestamp);
                }
            }, Event.class));

            this.myArrregategingState = getRuntimeContext().getAggregatingState(new AggregatingStateDescriptor<Event, Long, String>("myAggregatingState", new AggregateFunction<Event, Long, String>() {
                @Override
                public Long createAccumulator() {
                    return 0L;
                }

                @Override
                public Long add(Event value, Long accumulator) {
                    return accumulator + 1L;
                }

                @Override
                public String getResult(Long accumulator) {
                    return "count :" + accumulator;
                }

                @Override
                public Long merge(Long a, Long b) {
                    return null;
                }
            }, Long.class));
        }

        @Override
        public void flatMap(Event value, Collector<String> out) throws Exception {

            if (value.user.equalsIgnoreCase("Mary")){
                this.myArrregategingState.add(value);
            }

            System.out.println(value.user + " " + this.myArrregategingState.get());

        }
    }
}
